package cn.getech.data.development.utils

import java.sql.{Timestamp, Types}

import cn.getech.data.development.bean.jdbc.JDBCFieldBean
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.types.DataType
import org.apache.kudu.client.PartialRow

/**
 * 数据类型转换
 */
class JDBCSQLTypeUtils {

  def parquet_type2String(class_type: String): String = {
    class_type.toLowerCase match {
      case "bigint" => "long"
      case _ => class_type.toLowerCase
    }
  }

  // kudu
  def kudu_type2DataType(row: PartialRow, tuple: JDBCFieldBean, value: String): Unit = {
    tuple.class_type.toUpperCase() match {
      case "STRING" | "VARCHAR" =>
        row.addString(tuple.field_name, value)
      case "DECIMAL" =>
        row.addDecimal(tuple.field_name, java.math.BigDecimal.valueOf(value.toDouble))
      case "INT" | "INT32" =>
        row.addInt(tuple.field_name, value.toInt)
      case "TIMESTAMP" | "unixtime_micros" =>
        row.addTimestamp(tuple.field_name, Timestamp.valueOf(value.toString))
      case "LONG" | "BIGINT" | "INT64" =>
        row.addLong(tuple.field_name, value.toLong)
      case "DOUBLE" =>
        row.addDouble(tuple.field_name, value.toDouble)
      case "BOOL" | "BOOLEAN" =>
        row.addBoolean(tuple.field_name, value.toBoolean)
      case "INT8" | "BYTE" | "TINYINT" =>
        row.addByte(tuple.field_name, value.toByte)
      case "INT16" | "SHORT" | "SMALLINT" =>
        row.addShort(tuple.field_name, value.toShort)
      case "FLOAT" =>
        row.addFloat(tuple.field_name, value.toFloat)
      case "n" =>
        row.setNull(tuple.field_name)
      case _ =>
        throw new UnsupportedOperationException("Unknown type " + tuple.class_type)
    }
  }



  // jdbc
  def type2DataType(s: String): DataType = {
    s.toUpperCase match {
      case "INT" => DataTypes.INT()
      case "TINYINT" => DataTypes.TINYINT()
      case "SMALLINT" => DataTypes.SMALLINT
      case "INTEGER" | "BIGINT" => DataTypes.BIGINT()
      case "FLOAT" => DataTypes.FLOAT
      case "DOUBLE" => DataTypes.DOUBLE
      case "CHAR" | "STRING" | "VARCHAR" | "LONGVARCHAR" => DataTypes.STRING()
      case "DATE" => DataTypes.DATE
      case "TIME" => DataTypes.TIME
      case "TIMESTAMP" => DataTypes.TIMESTAMP
      case "VARBINARY" | "BYTES" => DataTypes.BYTES
      case "NULL" => DataTypes.NULL
      case "BOOLEAN" => DataTypes.BOOLEAN
      case _ =>
        throw new UnsupportedOperationException("Unknown type " + s)
    }
  }

  // jdbc
  def type2Int(s: String): Int = {
    s.toUpperCase match {
      case "BIT" => Types.BIT
      case "TINYINT" => Types.TINYINT
      case "SMALLINT" => Types.SMALLINT
      case "INTEGER" => Types.INTEGER
      case "BIGINT" => Types.BIGINT
      case "FLOAT" => Types.FLOAT
      case "REAL" => Types.REAL
      case "DOUBLE" => Types.DOUBLE
      case "NUMERIC" => Types.NUMERIC
      case "DECIMAL" => Types.DECIMAL
      case "CHAR" => Types.CHAR
      case "VARCHAR" | "STRING" => Types.VARCHAR
      case "LONGVARCHAR" => Types.LONGVARCHAR
      case "DATE" => Types.DATE
      case "TIME" => Types.TIME
      case "TIMESTAMP" => Types.TIMESTAMP
      case "VARBINARY" => Types.VARBINARY
      case "LONGVARBINARY" => Types.LONGVARBINARY
      case "NULL" => Types.NULL
      case "OTHER" => Types.OTHER
      case "JAVA_OBJECT" => Types.JAVA_OBJECT
      case "DISTINCT" => Types.DISTINCT
      case "STRUCT" => Types.STRUCT
      case "ARRAY" => Types.ARRAY
      case "BLOB" => Types.BLOB
      case "CLOB" => Types.CLOB
      case "REF" => Types.REF
      case "DATALINK" => Types.DATALINK
      case "BOOLEAN" => Types.BOOLEAN
      case "ROWID" => Types.ROWID
      case "NCHAR" => Types.NCHAR
      case "NVARCHAR" => Types.NVARCHAR
      case "LONGNVARCHAR" => Types.LONGNVARCHAR
      case "NCLOB" => Types.NCLOB
      case "SQLXML" => Types.SQLXML
      case "REF_CURSOR" => Types.REF_CURSOR
      case "TIME_WITH_TIMEZONE" => Types.TIME_WITH_TIMEZONE
      case "TIMESTAMP_WITH_TIMEZONE" => Types.TIMESTAMP_WITH_TIMEZONE
      case _ =>
        throw new UnsupportedOperationException("Unknown type " + s)
    }
  }

  def setRowField(): Unit = {

  }
}
